﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Text.RegularExpressions;
using System.IO;
using System.Diagnostics;
using Microsoft.WindowsAzure;
using Microsoft.WindowsAzure.Storage;
using Microsoft.WindowsAzure.ServiceRuntime;
using Microsoft.WindowsAzure.Diagnostics;
using System.Threading;
using Microsoft.WindowsAzure.Storage.Queue;
using Microsoft.WindowsAzure.Storage.Queue.Protocol;
using Microsoft.WindowsAzure.Storage.Blob;
using System.Runtime.Serialization.Formatters.Binary;
namespace cloudMPI
{
    /// <summary>
    /// I know the name does not give away my purpose, but I am used to create a Task. You know pack the data you will need to process and label it with a name.
    /// </summary>
    public class cMPI_Message
    {
        public string data { set; get; }
        public string datatype { set; get; }
        public string count { set; get; }
        private string _tag;
        public string source { set; get; }
        public string dest { set; get; }
        public cMPI_Message()
        {
        }
        public cMPI_Message(string tag)
        {
            _tag = tag;
        }
    }

    public class cMPI
    {
        private string _connectionString = "";
        private string _workerRoleName = "";
        private Dictionary<string,Type> MPI_Types = new Dictionary<string,Type>();
        private Dictionary<string, int> COMM_SIZES = new Dictionary<string, int>();
        private int MPI_RANK { set; get; }

        public cMPI(string connectionString, string workerRoleName)
        {
            _connectionString = connectionString;
            _workerRoleName = workerRoleName;
            setDataTypes();
        }

        public void Init(int slaveIDFrom, int slaveIDTo, string MPI_COMM="default")
        {
            TableHelper table = new TableHelper(_connectionString);
            TableData commSizeInfo = new TableData(DataConstants.MPI_COMM_SIZE_PART,DataConstants.MPI_COMM_SIZE_ROW);

            //table.CreateTable(MPI_COMM);
            commSizeInfo.Data = (slaveIDFrom - slaveIDTo + 1).ToString() + "|" + slaveIDFrom.ToString();
            //table.InsertEntity(MPI_COMM, commSizeInfo);

            /* Create the queues for this communicator */
            if (!createQueues(MPI_COMM, slaveIDFrom, slaveIDTo))
            {
                Exception rptError = new Exception("I can't setup queues. Have you created an internal endpoint for all roles?");
                throw rptError;
            }
        }


        private bool createQueues(string MPI_COMM, int slaveIDFrom, int slaveIDTo)
        {
            string queueName = "";
            int numberOfWorkers = RoleEnvironment.Roles[_workerRoleName].Instances.Count;
            /* Catch errors in Queue creation for workers */
            try
            {
                /* Create Queues for all slaves */
                QueueHelper workerQueue = new QueueHelper(_connectionString);
                for (int id = slaveIDFrom; id <= slaveIDTo; id++)
                {
                    /* QueueName is tied to the worker ID */
                    queueName = MPI_COMM + DataConstants.MPI_QUEUE + id.ToString();
                    if (!workerQueue.CreateQueue(queueName))
                    {
                        Exception creationError = new Exception("Cannot create the queue. Have you created an endpoint? If yes, may be you are trying too soon, wait a minute and try again.");
                    }
                }

                /* Create Queue for the master */
                /* QueueName is tied to the worker ID */
                queueName = MPI_COMM + DataConstants.MPI_MASTER_QUEUE;
                if (!workerQueue.CreateQueue(queueName))
                {
                    Exception creationError = new Exception("Cannot create the queue. Have you created an endpoint? If yes, may be you are trying too soon, wait a minute and try again.");
                }
            }
            catch (Exception ex)
            {
                /* Queue might have thrown an exception as well */
                System.Diagnostics.Trace.WriteLine("Error during Queue creation:" + ex.Message);
                throw ex;
            }
            return true;
        }

        public int Send(cMPI_Message message, string MPI_COMM="default", bool toMaster=false)
        {
            QueueHelper queue = new QueueHelper(_connectionString);
            try
            {
                /* If data cannot fit in one message store it in blob */
                string strMessage = messageToString(message);
                string msgID = Guid.NewGuid().ToString();
                if ((strMessage.Length + msgID.Length) > DataConstants.MAX_MSG_SIZE)
                {
                    BlobHelper blob = new BlobHelper(_connectionString);

                    if (!blob.PutBlob(DataConstants.MESSAGE_BLOB_CONTAINER, msgID, strMessage))
                    {
                        Exception ex = new Exception("Error sending message");
                        throw ex;
                    }
                    /* Put this blob's id in the bag queue.*/
                    if (!queue.PutMessage(getQueueName(MPI_COMM, message.dest, toMaster), new CloudQueueMessage("b" + msgID)))
                        {
                            Thread.Sleep(1000);
                            return 0;
                        }
                }
                else
                {
                    /* The message is not big enough, Store it in Queue itself.*/
                    if (!queue.PutMessage(getQueueName(MPI_COMM, message.dest, toMaster), new CloudQueueMessage("q" + msgID + "|" + strMessage)))
                    {
                        Thread.Sleep(1000);
                        return 0;
                    }
                }
                /* We are done adding the task. Put a message in task count queue that represents original task count*/
            }
            catch (Exception ex)
            {
                throw ex;
            }
                return 1; 
        }

        public int BroadCast(cMPI_Message message, string MPI_COMM = "default", bool toMaster = false)
        {
            QueueHelper queue = new QueueHelper(_connectionString);
            try
            {
                TableHelper table = new TableHelper(_connectionString);
                TableData slaveIDs = new TableData();
                table.GetEntity(MPI_COMM, DataConstants.MPI_COMM_SIZE_PART, DataConstants.MPI_COMM_SIZE_ROW, out slaveIDs);
                int idFrom;
                int.TryParse(slaveIDs.Data.Split('|')[1], out idFrom);
                int commSize;
                int.TryParse(slaveIDs.Data.Split('|')[0], out commSize);
                int idTo = idFrom + commSize;
                for (int dest = idFrom; dest < idTo; dest++)
                {
                    /* If data cannot fit in one message store it in blob */
                    string strMessage = messageToString(message);
                    string msgID = Guid.NewGuid().ToString();
                    if ((strMessage.Length + msgID.Length) > DataConstants.MAX_MSG_SIZE)
                    {
                        BlobHelper blob = new BlobHelper(_connectionString);

                        if (!blob.PutBlob(DataConstants.MESSAGE_BLOB_CONTAINER, msgID, strMessage))
                        {
                            Exception ex = new Exception("Error sending message");
                            throw ex;
                        }
                        /* Put this blob's id in the bag queue.*/
                        if (!queue.PutMessage(getQueueName(MPI_COMM, dest.ToString(), toMaster), new CloudQueueMessage("b" + msgID)))
                        {
                            Thread.Sleep(1000);
                            return 0;
                        }
                    }
                    else
                    {
                        /* The message is not big enough, Store it in Queue itself.*/
                        if (!queue.PutMessage(getQueueName(MPI_COMM, dest.ToString(), toMaster), new CloudQueueMessage("q" + msgID + "|" + strMessage)))
                        {
                            Thread.Sleep(1000);
                            return 0;
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
            return 1;
        }

        public bool Recv(out cMPI_Message data, string destination, string MPI_COMM="default", bool fromMaster=false)
        {
            int msgCount = 0;
            string message;
            CloudQueueMessage _currentMsg;

            QueueHelper queue = new QueueHelper(_connectionString);
            /* Find the queue and see how many messages are there */
            string myQueueName = getQueueName(MPI_COMM, destination, fromMaster);
            if (!queue.GetApproxMessageCount(myQueueName, out msgCount))
            {
                Thread.Sleep(1000);
                data = null;
                return false;
            }

            /* If the queue does not have any message return false */
            if (msgCount == 0)
            {
                data = null;
                /* The worker can throttle us so wait for a second */
                Thread.Sleep(1000);
                return false;
            }

            /* so we do have a message. Let's read it from the  queue*/
            /* We will need to delete this message on successful processing so 
             * keep it in this object's currentMsg variable */
            if (!queue.GetMessage(myQueueName, out _currentMsg))
            {
                Thread.Sleep(1000);
                data = null;
                return false;
            }

            /* We got the message. Check if the data is in the message itself or in the blob */
            message = _currentMsg.AsString;
            data = new cMPI_Message();
            if (message[0] == 'q')
            {
                /* The message is in the Queue itself. Read it and store it in task's data member */
                string msg = message.Substring(1);
                //data.taskID = msg.Split(new Char[] { '|' })[0];
                data = stringToMessage(msg.Split(new Char[] { '|' })[1]);
            }
            else
            {
                /* The data is not in the queue. Read the blob as a stream and get data */
                string blobID = message.Substring(1);
                BlobHelper blob = new BlobHelper(_connectionString);
                string blobContent;
                blob.GetBlob(DataConstants.MESSAGE_BLOB_CONTAINER, blobID, out blobContent);
                data.data = blobContent;
            }
            queue.DeleteMessage(myQueueName, _currentMsg);
            return true;
        }

        public bool RecvMult(out List<cMPI_Message> data, int messageCount, string destination, string MPI_COMM = "default", bool fromMaster = false)
        {
            int msgCount = 0;
            string message;
            
            QueueHelper queue = new QueueHelper(_connectionString);
            /* Find the queue and see how many messages are there */
            string myQueueName = getQueueName(MPI_COMM, destination, fromMaster);
            
            if (messageCount > 32) messageCount = 32;

            if (!queue.GetApproxMessageCount(myQueueName, out msgCount))
            {
                Thread.Sleep(1000);
                data = null;
                return false;
            }

            /* If the queue does not have any message return false */
            if (msgCount == 0)
            {
                data = null;
                /* The worker can throttle us so wait for a second */
                Thread.Sleep(1000);
                return false;
            }
            /* If there are less messages than asked for, return everything */
            if (msgCount < messageCount) messageCount = msgCount;

            /* so we do have a message. Let's read it from the  queue*/
            /* We will need to delete this message on successful processing so 
             * keep it in this object's currentMsg variable */
            IEnumerable<CloudQueueMessage> msgs = new List<CloudQueueMessage>();
            if (!queue.GetMessage(myQueueName, messageCount, ref msgs))
            {
                Thread.Sleep(1000);
                data = null;
                return false;
            }
            data = new List<cMPI_Message>();
            List<cMPI_Message> output = new List<cMPI_Message>();
            /* We got the message. Check if the data is in the message itself or in the blob */
            var iterator = msgs.GetEnumerator();
            while (iterator.MoveNext())
            {
                message = iterator.Current.AsString;
                if (message[0] == 'q')
                {
                    /* The message is in the Queue itself. Read it and store it in task's data member */
                    string msg = message.Substring(1);
                    //data.taskID = msg.Split(new Char[] { '|' })[0];
                    output.Add(stringToMessage(msg.Split(new Char[] { '|' })[1]));
                }
                else
                {
                    /* The data is not in the queue. Read the blob as a stream and get data */
                    string blobID = message.Substring(1);
                    BlobHelper blob = new BlobHelper(_connectionString);
                    string blobContent;
                    blob.GetBlob(DataConstants.MESSAGE_BLOB_CONTAINER, blobID, out blobContent);
                    cMPI_Message blobMsg = new cMPI_Message();
                    blobMsg.data = blobContent;
                    output.Add(blobMsg);
                }
                queue.DeleteMessage(myQueueName, iterator.Current);
            }
            data = output;
            return true;
        }
        public void cMPI_Comm_Size(out int Size, string MPI_COMM = "default")
        {
            try
            {
                TableHelper table = new TableHelper(_connectionString);
                TableData COMM_SIZE_INFO = new TableData();
                table.GetEntity(MPI_COMM, DataConstants.MPI_COMM_SIZE_PART, DataConstants.MPI_COMM_SIZE_ROW, out COMM_SIZE_INFO);
                if (!int.TryParse(COMM_SIZE_INFO.Data, out Size))
                {
                    Exception ex = new Exception("Cannot fetch data");
                    throw ex;
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }

        public int getRank()
        {
            string instanceId = RoleEnvironment.CurrentRoleInstance.Id;
            int instanceIndex = 0;
            if (!int.TryParse(instanceId.Substring(instanceId.LastIndexOf(".") + 1), out instanceIndex)) // On cloud.
            {
                int.TryParse(instanceId.Substring(instanceId.LastIndexOf("_") + 1), out instanceIndex); // On compute emulator.
            }

            return instanceIndex ;
        }

        void setDataTypes()
        {
            MPI_Types.Add("MPI_CHAR", typeof(Char));
            MPI_Types.Add("MPI_SHORT", typeof(Int16));
            MPI_Types.Add("MPI_INT", typeof(Int32));
            MPI_Types.Add("MPI_LONG", typeof(Int64));
            MPI_Types.Add("MPI_BYTE", typeof(Byte));
            MPI_Types.Add("MPI_FLOAT", typeof(Single));
            MPI_Types.Add("MPI_DOUBLE", typeof(Double));
            MPI_Types.Add("MPI_UNSIGNED", typeof(UInt32));
            MPI_Types.Add("MPI_UNSIGNED_SHORT", typeof(UInt16));
            MPI_Types.Add("MPI_UNSIGNED_LONG", typeof(UInt64));
            MPI_Types.Add("MPI_LONG_DOUBLE", typeof(Decimal));
        }
        
        private string messageToString(cMPI_Message message)
        {
            string returnMessage = "";
            returnMessage += message.data + "}{" + message.datatype + "}{" + message.count + "}{" + message.source;
            return returnMessage;
        }

        private cMPI_Message stringToMessage(string message)
        {
            cMPI_Message msg = new cMPI_Message();
            string[] separators = new string[] {"}{"};
            string[] returnMessage = message.Split(separators, StringSplitOptions.None);
            msg.data = returnMessage[0];
            msg.datatype = returnMessage[1];
            msg.count = returnMessage[2]; 
            msg.source = returnMessage[3];
            return msg;
        }

        private void getWorkerID(out string workerID)
        {
            string instanceId = RoleEnvironment.CurrentRoleInstance.Id;
            int instanceIndex = 0;
            if (!int.TryParse(instanceId.Substring(instanceId.LastIndexOf(".") + 1), out instanceIndex)) // On cloud.
            {
                int.TryParse(instanceId.Substring(instanceId.LastIndexOf("_") + 1), out instanceIndex); // On compute emulator.
            }
            workerID = instanceIndex.ToString();
        }

        private static string getQueueName(string MPI_COMM, string dest, bool isMaster)
        {
            if (!isMaster)
            {
                return (MPI_COMM + DataConstants.MPI_QUEUE + dest);
            }
            else
            {
                return MPI_COMM + DataConstants.MPI_MASTER_QUEUE;
            }
        }
    }
}
